package com.innodealing.controller;

import com.innodealing.service.KafkaProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;

/**
 * Kafka测试控制器
 * 提供简洁的HTTP接口来批量发送测试消息
 */
@RestController
@RequestMapping("/kafka/test")
public class KafkaProducerController {

    @Autowired
    private KafkaProducerService producer;

    /**
     * 批量发送消息（核心接口）
     *
     * @param topic 主题名称
     * @param partition 分区号（可选）
     * @param count 消息数量
     * @param errorPositions 错误位置（逗号分隔，如："2,5,8"）
     * @param prefix 消息前缀
     */
    @PostMapping("/send-batch")
    public Map<String, Object> sendBatch(
            @RequestParam String topic,
            @RequestParam(required = false) Integer partition,
            @RequestParam(defaultValue = "10") int count,
            @RequestParam(required = false) String errorPositions,
            @RequestParam(defaultValue = "test") String prefix) {
        
        List<Integer> errors = parseErrorPositions(errorPositions);
        List<SendResult<String, String>> results = producer.sendBatchWithErrors(
                topic, partition, count, errors, prefix);
        
        producer.printSendStatistics(results);
        
        Map<String, Object> result = new HashMap<>();
        result.put("success", true);
        result.put("message", "批量发送完成");
        result.put("topic", topic);
        result.put("partition", partition != null ? partition : "auto");
        result.put("totalCount", count);
        result.put("errorPositions", errors);
        result.put("sentCount", results.size());
        return result;
    }

    /**
     * 解析错误位置字符串
     */
    private List<Integer> parseErrorPositions(String errorPositions) {
        if (errorPositions == null || errorPositions.trim().isEmpty()) {
            return Collections.emptyList();
        }
        
        try {
            List<Integer> result = new ArrayList<>();
            for (String pos : errorPositions.split(",")) {
                result.add(Integer.parseInt(pos.trim()));
            }
            return result;
        } catch (NumberFormatException e) {
            return Collections.emptyList();
        }
    }
}
